package org.multi.core;


import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.BinaryOperator;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
import java.util.function.ToDoubleFunction;
import java.util.function.ToIntFunction;
import java.util.function.ToLongFunction;
import java.util.function.UnaryOperator;
import java.util.stream.Collector;
import org.multi.core.function.BiIntConsumer;
import org.multi.core.function.BiIntFunction;
import org.multi.core.function.TripleConsumer;
import org.multi.core.state.IntState;
import org.multi.core.state.LongState;

public interface Multi<T> {

    static void stop() {
        StopException.stop();
    }

    static <T> Multi<T> empty() {
        return consumer -> {
        };
    }

    static <T> Multi<T> of(T item) {
        return consumer -> consumer.accept(item);
    }

    @SafeVarargs
    static <T> Multi<T> of(T... items) {
        return consumer -> {
            for (T item : items) {
                consumer.accept(item);
            }
        };
    }

    static <T> Multi<T> of(Iterable<T> items) {
        return consumer -> {
            for (T item : items) {
                consumer.accept(item);
            }
        };
    }

    static <T> Multi<T> create(Consumer<Consumer<T>> consumer) {
        return consumer::accept;
    }

    static <T> Multi<T> iterate(T seed, UnaryOperator<T> next) {
        return consumer -> {
            T item = seed;
            consumer.accept(item);
            while (true) {
                consumer.accept(item = next.apply(item));
            }
        };
    }

    static IntMulti range(int start, int end) {
        return IntMulti.range(start, end);
    }

    static IntMulti rangeClosed(int start, int end) {
        return IntMulti.rangeClosed(start, end);
    }

    static CharMulti range(char start, char end) {
        return CharMulti.range(start, end);
    }

    static CharMulti rangeClosed(char start, char end) {
        return CharMulti.rangeClosed(start, end);
    }

    static <T> Multi<T> iterate(T seed1, T seed2, BinaryOperator<T> next) {
        return consumer -> {
            T a = seed1;
            T b = seed2;
            consumer.accept(a);
            consumer.accept(b);
            while (true) {
                consumer.accept(b = next.apply(a, a = b));
            }
        };
    }

    static <T> Multi<T> repeat(T item, long count) {
        return consumer -> {
            for (long i = 0; i < count; i++) {
                consumer.accept(item);
            }
        };
    }

    default void consume(Consumer<T>[] consumers) {
        consume(item -> {
            for (Consumer<T> consumer : consumers) {
                consumer.accept(item);
            }
        });
    }

    default void consume(long from, long to, Consumer<T> consumer) {
        LongState state = new LongState(0);
        drop(from).consumeOn(item -> {
            if (state.value < to) {
                state.value++;
                consumer.accept(item);
            } else {
                stop();
            }
        });
    }

    void consume(Consumer<T> onItem);

    default Multi<T> onFailure(Consumer<Throwable> onFailure) {
        return consumer -> consume(consumer, onFailure);
    }

    default void consume(Consumer<T> onItem, Consumer<Throwable> onFailure) {
        try {
            consume(onItem);
        } catch (Throwable error) {
            onFailure.accept(error);
        }
    }

    default void consume(int from, Consumer<T> headConsumer, int to, Consumer<T> otherConsumer) {
        if (from < 0 || to < 0 || from > to) {
            consume(headConsumer);
        }
        IntState state = new IntState(from);
        consume(item -> {
            if (state.value < to) {
                state.value++;
                headConsumer.accept(item);
            } else {
                otherConsumer.accept(item);
            }
        });
    }

    default void consume(Consumer<T> headConsumer, int to, Consumer<T> otherConsumer) {
        consume(0, headConsumer, to, otherConsumer);
    }

    default void consumeOn(Consumer<T> consumer) {
        try {
            consume(consumer);
        } catch (StopException ignored) {
        }
    }

    default void consumeWithIndex(BiIntConsumer<T> consumer) {
        int[] state = {0};
        consume(item -> consumer.accept(state[0]++, item));
    }

    default void forEach(Consumer<T> consumer) {
        consume(consumer);
    }

    default <R> Multi<R> map(Function<T, ? extends R> mapper) {
        return consumer -> consume(item -> consumer.accept(mapper.apply(item)));
    }

    default <R> Multi<R> mapMulti(BiConsumer<T, Consumer<R>> function) {
        return consumer -> consume(item -> function.accept(item, consumer));
    }

    default <R> Multi<R> mapIndexed(BiIntFunction<T, R> mapper) {
        return consumer -> consumeWithIndex((index, item) -> consumer.accept(mapper.apply(index, item)));
    }

    default Multi<T> filter(Predicate<T> predicate) {
        return consumer -> consume(item -> {
            if (predicate.test(item)) {
                consumer.accept(item);
            }
        });
    }

    default Multi<T> head() {
        return take(1);
    }

    default Multi<T> tail() {
        return drop(1);
    }

    default boolean isEmpty() {
        boolean[] state = {true};
        consumeOn(item -> {
            state[0] = false;
            stop();
        });
        return state[0];
    }

    default Multi<T> flatMap(Function<T, Multi<T>> mapper) {
        return consumer -> consume(item -> mapper.apply(item).consume(consumer));
    }

    default Multi<T> append(T item) {
        return consumer -> {
            consume(consumer);
            consumer.accept(item);
        };
    }

    @SuppressWarnings("unchecked")
    default Multi<T> append(T... items) {
        return consumer -> {
            consume(consumer);
            for (T item : items) {
                consumer.accept(item);
            }
        };
    }

    default Multi<T> append(Iterable<T> items) {
        return consumer -> {
            consume(consumer);
            for (T item : items) {
                consumer.accept(item);
            }
        };
    }

    default Multi<T> take(long count) {
        return consumer -> {
            long[] state = {count};
            consumeOn(item -> {
                consumer.accept(item);
                if (--state[0] == 0) {
                    stop();
                }
            });
        };
    }

    default Multi<T> takeWhile(Predicate<T> test) {
        return consumer -> {
            consumeOn(item -> {
                if (test.test(item)) {
                    consumer.accept(item);
                } else {
                    stop();
                }
            });
        };
    }

    default Multi<T> drop(long count) {
        return consumer -> {
            long[] state = {count};
            consume(item -> {
                if (state[0] > 0) {
                    state[0]--;
                } else {
                    consumer.accept(item);
                }
            });
        };
    }

    default Multi<T> dropWhile(Predicate<T> test) {
        return consumer -> {
            boolean[] state = {true};
            consume(item -> {
                if (state[0] && test.test(item)) {
                    return;
                }
                state[0] = false;
                consumer.accept(item);
            });
        };
    }

    default Multi<T> peek(Consumer<T> action) {
        return consumer -> consume(consumer.andThen(action));
    }

    default <E, R> Multi<R> zip(Iterable<E> iterable, BiFunction<T, E, R> zipper) {
        return consumer -> consume(item -> {
            for (E element : iterable) {
                consumer.accept(zipper.apply(item, element));
            }
        });
    }

    default boolean anyMatch(Predicate<T> test) {
        boolean[] state = {false};
        consumeOn(item -> {
            if (test.test(item)) {
                state[0] = true;
                stop();
            }
        });
        return state[0];
    }

    default boolean allMatch(Predicate<T> test) {
        return !anyMatch(test.negate());
    }

    default boolean noneMatch(Predicate<T> test) {
        return anyMatch(test.negate());
    }

    default <A, R> R reduce(Reducer<T, A, R> reducer) {
        Supplier<A> supplier = reducer.supplier();
        BiConsumer<A, T> accumulator = reducer.accumulator();
        A result = supplier.get();
        consume(item -> accumulator.accept(result, item));
        Function<A, R> finisher = reducer.finisher();
        return finisher.apply(result);
    }

    default <R> R reduce(R des, BiConsumer<R, T> accumulator) {
        consume(item -> accumulator.accept(des, item));
        return des;
    }

    default <A, R> R reduce(Collector<T, A, R> collector) {
        return reduce(Reducer.of(collector));
    }

    default <A, R> Multi<R> gather(Gatherer<T, A, R> gatherer) {
        Supplier<A> supplier = gatherer.supplier();
        A state = supplier.get();
        TripleConsumer<A, T, Consumer<R>> accumulator = gatherer.accumulator();
        BiConsumer<A, Consumer<R>> finisher = gatherer.finisher();
        return consumer -> {
            consume(item -> accumulator.accept(state, item, consumer));
            finisher.accept(state, consumer);
        };
    }

    default List<T> toList() {
        return reduce(new ArrayList<>(10), List::add);
    }

    default List<T> toList(int initialCapacity) {
        return reduce(Reducer.toList(initialCapacity));
    }

    default Set<T> toSet() {
        return reduce(Reducer.toSet());
    }

    default Set<T> toSet(int initialCapacity) {
        return reduce(Reducer.toSet(initialCapacity));
    }

    default <R extends Collection<T>> R collect(Supplier<R> supplier) {
        return reduce(Reducer.collect(supplier));
    }

    default List<T> sort() {
        return reduce(Reducer.sort());
    }

    default List<T> sort(Comparator<T> comparator) {
        return reduce(Reducer.sort(comparator));
    }

    default List<T> sortDesc(Comparator<T> comparator) {
        return reduce(Reducer.sortDesc(comparator));
    }

    default <R extends Comparable<R>> List<T> sort(Function<T, R> function) {
        return sort(Comparator.comparing(function));
    }

    default <R extends Comparable<R>> List<T> sortDesc(Function<T, R> function) {
        return sort(Comparator.comparing(function).reversed());
    }

    default long count() {
        return reduce(Reducer.count());
    }

    default Multi<T> distinct() {
        return reduce(Reducer.distinct());
    }

    default long count(Predicate<T> predicate) {
        return reduce(Reducer.count(predicate));
    }

    default double sum(ToDoubleFunction<T> mapper) {
        return reduce(Reducer.sum(mapper));
    }

    default int sumInt(ToIntFunction<T> mapper) {
        return reduce(Reducer.sumInt(mapper));
    }

    default long sumLong(ToLongFunction<T> mapper) {
        return reduce(Reducer.sumLong(mapper));
    }

    default String join(String delimiter) {
        return reduce(Reducer.join(delimiter));
    }

    default String join(String delimiter, String prefix, String suffix) {
        return reduce(Reducer.join(delimiter, prefix, suffix, Object::toString));
    }

    default IntMulti mapToInt(ToIntFunction<T> function) {
        return consumer -> consume(item -> consumer.accept(function.applyAsInt(item)));
    }

    default void println() {
        consume(System.out::println);
    }

}
